import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.*;

public class Acceptor extends ServerThread {

    private final ServerSocketChannel serverSocketChannel;
    private final Server server;
    //subscriber线程集合，子线程集合
    private final Collection<PollerIO> ioThreads;
    //方便获取 ioThreads集合中的元素
    private Iterator<PollerIO> ioIterator;

    public Acceptor(String name, Server server, Set<PollerIO> ioThreads) throws IOException {
        super(name);
        this.server = server;
        //主reactor 开启一个serverSocketChannel  底层：epoll_creat()
        this.serverSocketChannel = ServerSocketChannel.open();
        //绑定端口
        this.serverSocketChannel.bind(new InetSocketAddress(server.getPort()));
        //设置为非阻塞
        this.serverSocketChannel.configureBlocking(false);
        //register to selector 开启 accept事件,底层 epoll_ctl()
        this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);

        //创建一个不可变的集合
        this.ioThreads = Collections.unmodifiableList(new ArrayList<>(ioThreads));
        this.ioIterator = this.ioThreads.iterator();
    }

    /**
     * 主reactor 的run方法
     * 主要处理accept的事件
     */
    @Override
    public void run() {
        //服务没有停止，并且 serverSocketChannel没有关闭
        while (!server.stopped && !serverSocketChannel.socket().isClosed()) {
            try {
                this.selector.select();
                Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    if (!key.isValid()) {
                        continue;
                    }
                    //accept事件
                    if (key.isAcceptable()) {
                        //执行业务逻辑
                        doAccept(key);
                    }
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * //get pollerIO,  register to selector
     * accept事件，获取到新的客户端连接到 的时候，需要注册到selector
     */
    private void doAccept(SelectionKey key) throws IOException {
        //SocketChannel socketChannel = (SocketChannel)key.channel();
        SocketChannel socketChannel = this.serverSocketChannel.accept();
        socketChannel.configureBlocking(false);
        //如果迭代器中没有元素了，则重新初始化迭代器
        if (!ioIterator.hasNext()) {
            this.ioIterator = this.ioThreads.iterator();
        }
        PollerIO pollerIO = this.ioIterator.next();

        //主线程的accept事件
        pollerIO.addAcceptedConnection(socketChannel);
    }
}
